package com.shujia.rec.funaction

import com.shujia.rec.entry.CaseClass.{CountEntry, LogEntry}
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * 统计一个窗口中同一个商品的数量
  *
  */
class CountProcessWindowFunction extends ProcessWindowFunction[LogEntry, CountEntry, Int, TimeWindow] {


  override def process(key: Int, context: Context, elements: Iterable[LogEntry], out: Collector[CountEntry]): Unit = {

    val start = context.window.getStart
    val end = context.window.getEnd

    //统计一个窗口中同一个商品的数量
    val count = elements.map(log => 1).sum

    //返回
    out.collect(CountEntry(key, count, start, end))
  }
}
